package demo06

/**
 * @author xiaoZhao
 * @date 2022/7/19
 * @describe 复杂版的wordCount
 */
object test18_ComplexWordCount {
  def main(args: Array[String]): Unit = {
    // 创建单词集合
    val tupleList = List(
      ("hello", 1),
      ("hello world", 3),
      ("hello scala", 4),
      ("hello scala spark", 7),
      ("hello scala spark flink",5)
    )

    // 解法一：直接展开为普通版本
    val stringList = tupleList.map(
      kv => {
        (kv._1.trim + " ") * kv._2
      }
    )

    val wordCountList = stringList
      .flatMap(_.split(" "))
      .groupBy(word => word)
      .map(kv => (kv._1, kv._2.length))
      .toList
      .sortBy(_._2)(Ordering[Int].reverse)
      .take(3)

    println(wordCountList)

    // 解法二：基于预统计的结果进行转换
    // 将字符串打散为单词
    val preCountlist=tupleList.flatMap(
      tuple => {
        val tuples = tuple._1.split(" ")
          .map(word => (word, tuple._2))
        tuples
      }
    )

    // 对二元组进行单词进行分组
    val preCountMap = preCountlist.groupBy(tuple => tuple._1)

    // 叠加每个单词统计的个数值
    val countMap = preCountMap.mapValues(
      tupleList => tupleList.map(_._2).sum
    )

    // 转换成List 排序输出
    val countList = countMap.toList
      .sortWith(_._2 > _._2)
      .take(3)

    println(countList)
  }
}
